package com.tydic;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.StreamTableEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.sinks.CsvTableSink;
import org.apache.flink.table.sinks.TableSink;
import org.apache.flink.table.sources.CsvTableSource;
import org.apache.flink.table.sources.TableSource;

public class CsvSourceToCsvSink {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);

        String[] sourceFieldName = {"xm", "age", "address", "time"};
        TypeInformation[] sourceFieldTypes = {Types.STRING(), Types.INT(), Types.STRING(), Types.STRING()};
        TableSource csvSource = new CsvTableSource("/flink/data/01.csv", sourceFieldName, sourceFieldTypes);
        tableEnv.registerTableSource("CsvSourceTable", csvSource);

        TableSink csvSink = new CsvTableSink("/flink/result/01", "|");
        String[] sinkFieldNames = {"xm", "age", "address", "time"};
        TypeInformation[] sinkFieldTypes = {Types.STRING(), Types.INT(), Types.STRING(), Types.STRING()};
        tableEnv.registerTableSink("CsvSinkTable", sinkFieldNames, sinkFieldTypes, csvSink);

        Table result = tableEnv.scan("CsvSourceTable").select("age,address");
        result.writeToSink(csvSink);

        env.execute();

    }
}
